package com.person.server;

import com.person.server.grpcjava.PersonServiceGrpc;
import com.person.server.grpcjava.StreamRequest;
import com.person.server.grpcjava.StreamResponse;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;


/**
 * 详细介绍，grpc这四种请求方式
 * 1.客户端非流式，服务端非流式(客户端发送一次消息，服务端接收一次消息。服务端可响应一条消息，客户端接收一次响应消息）
 * 2.客户端非流式，服务端流式请求(客户端发送一次，服务端接收一次消息。 服务端可响应多条消息onNext，当服务端onCompleted时，客户端在迭代器一次收到多条响应消息）
 * 3.客户端流式，服务端非流式(客户端发送多次onNext，服务器接收多次消息。服务端可响应一条消息，客户端接收一次响应消息）
 * 4.客户端流式，服务端流式(客户端发送多次onNext，服务器接收多次消息onNext。客户端和服务端都无需onCompleted，服务端响应多条消息onNext，客户端接收多次响应消息）
 * <p>
 * 说明：流式即消息多次发送。非流式只能发送一次
 */
public class GrpcClientBiTalk2 {

    final CountDownLatch loginLatch = new CountDownLatch(1);
    //消息缓存
    Map<String, String> msgCache = new HashMap<String, String>();

    String clientId = "client002";
    /**
     * 模拟客户端先与服务器建立连接和登录、通知服务器可以开始持续牌价的场景、新增主动与服务器心跳场景
     * @param args
     */
    public static void main(String[] args) {

        GrpcClientBiTalk2 grpcClient = new GrpcClientBiTalk2();
        //连接
        ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost", 8899)
                .usePlaintext().build();
        PersonServiceGrpc.PersonServiceStub stub = PersonServiceGrpc.newStub(managedChannel);

        //获取requestStreamObserver
        StreamObserver<StreamRequest> requestStreamObserver = stub.biTalk(grpcClient.streamObserverImpl());

        //登录，新跳
        String loginResult = grpcClient.login(requestStreamObserver);
        System.out.println("登录结果：" + loginResult);

        //可以开始收牌价了
        grpcClient.getPriceMsg(requestStreamObserver);

        //心跳检测
        if (loginResult.contains("成功")) {
            //开始像服务端发心跳
            grpcClient.heart(requestStreamObserver);
        }



    }

    public void getPriceMsg(StreamObserver<StreamRequest> requestStreamObserver) {
        //开始登录
        requestStreamObserver.onNext(StreamRequest.newBuilder().setRequestInfo("okPrice").setClientId(clientId).build());
    }

    public String login(StreamObserver<StreamRequest> requestStreamObserver) {
        //开始登录
        requestStreamObserver.onNext(StreamRequest.newBuilder().setRequestInfo("login").setClientId(clientId).build());
        try {
            System.out.println("阻塞直到流返回");
            loginLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return msgCache.get(clientId);

    }

    public void heart(StreamObserver<StreamRequest> requestStreamObserver) {
        //开始登录
        ScheduledExecutorService timer = Executors.newScheduledThreadPool(1);
        timer.scheduleWithFixedDelay(() -> {
            //lamba表达式写一个线程
            try {
                requestStreamObserver.onNext(StreamRequest.newBuilder().setClientId(clientId).setRequestInfo("heartCheck").build());
                System.out.println("发送心跳");
            } catch (Exception e) {
                // 移除客户端
                timer.shutdownNow();
                e.getMessage();
            }
        }, 0, 3, TimeUnit.SECONDS);
    }


    public  StreamObserver<StreamResponse> streamObserverImpl(){

        //定义并实现StreamObserver接口streamObserverImpl为接口实现类，如果连接失败次处的streamObserverImpl不处理消息
        StreamObserver<StreamResponse> streamObserverImpl = new StreamObserver<StreamResponse>() {
            @Override
            public void onNext(StreamResponse streamResponse) {
                // 启动成功，连接已建立（之后可以把服务器断开，客户端需要一直重试，直到服务器重启成功）
                 if (streamResponse.getResponseInfo().contains("登录成功")) {
                    msgCache.put(clientId, streamResponse.getResponseInfo());
                    loginLatch.countDown();//登录成功暂停阻塞
                }else if (streamResponse.getResponseInfo().contains("心跳请求")) {
                     System.out.println(streamResponse.getResponseInfo());
                 } else if (streamResponse.getResponseInfo().contains("okPrice")) {
                     System.out.println(streamResponse.getResponseInfo());
                 }
            }

            //如果连接失败，调用retry重连，消息将有retry.streamObserverImpl重连成功对象处理
            @Override
            public void onError(Throwable throwable) {
                System.out.println(throwable.getMessage());
            }

            @Override
            public void onCompleted() {
                System.out.println("onComplated");
            }
        };
        return streamObserverImpl;
    }

}